package com.example.lc.myrxjavademo;

import android.app.Activity;
import android.database.Cursor;
import android.icu.text.AlphabeticIndex;
import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.Toast;

import com.jakewharton.rxbinding2.view.RxView;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import retrofit2.Call;

public class MainActivity extends Activity implements View.OnClickListener {
    private String TAG = "lc";
    private String APP_KEY = "86ab45ec310d097f";
    private Call call;
    private Button bt_content;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        initView();
    }

    private void initView() {
        bt_content = (Button) findViewById(R.id.bt_content);

        bt_content.setOnClickListener(this);

//        // ------------http://www.jianshu.com/p/464fa025229e 给初学者的RxJava2.0教程(一)------------
//        Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
//                // ObservableEmitter： Emitter是发射器的意思，那就很好猜了，这个就是用来发出事件的，它可以发出三种类型的事件，\
//                // 通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。
//                Log.d(TAG, "emit 1");
//                e.onNext(1);
//                Log.d(TAG, "emit 2");
//                e.onNext(2);
//                Log.d(TAG, "emit 3");
//                e.onNext(3);
//                Log.d(TAG, "emit complete");
//                e.onComplete();
////                e.onError(new NullPointerException("haha"));
//                Log.d(TAG, "emit 4");
//                e.onNext(4);
//
//            }
//        });
//        Observer<Integer> observer = new Observer<Integer>() {
//            private Disposable mDisposable;// Disposable的用处不止这些, 后面讲解到了线程的调度之后, 我们会发现它的重要性. 随着后续深入的讲解, 我们会在更多的地方发现它的身影.
//            private int i = 0;
//
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//                mDisposable = d;
//                Log.d(TAG, "subscribe");
//            }
//
//            @Override
//            public void onNext(@NonNull Integer integer) {
//                Log.d(TAG, "" + integer);
//                i++;
//                if (i == 2) {
//                    Log.d(TAG, "dispose");
//                    mDisposable.dispose();// 切断水管
//                    Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
//                }
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//                Log.d(TAG, "error" + e.toString());
//            }
//
//            @Override
//            public void onComplete() {
//                Log.d(TAG, "complete");
//            }
//        };
//        // 注意: 只有当上游和下游建立连接之后, 上游才会开始发送事件. 也就是调用了subscribe() 方法之后才开始发送事件.
//        integerObservable.subscribe(observer);


        // ------------http://www.jianshu.com/p/8818b98c44e2 给初学者的RxJava2.0教程(二)------------
        Log.d(TAG, Thread.currentThread().getName() + "---" + Thread.currentThread().getId());
        // 多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
        // 多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.

        // 在RxJava中, 已经内置了很多线程选项供我们选择, 例如有
        // Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
        // Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
        // Schedulers.newThread() 代表一个常规的新线程
        // AndroidSchedulers.mainThread() 代表Android的主线程

//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
//                Log.d(TAG, "emit 1");
//                emitter.onNext(1);
//                Log.d(TAG, "emit 2");
//                emitter.onNext(2);
//                Log.d(TAG, "emit 3");
//                emitter.onNext(3);
//                emitter.onComplete();
//            }
//        }).subscribeOn(Schedulers.newThread()
//        ).observeOn(AndroidSchedulers.mainThread()).doOnNext(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Exception {
//                Log.d(TAG, "After observeOn(mainThread), current thread is: " + Thread.currentThread().getName() + "   Integer=" + integer);
//                Thread.sleep(2000);
//            }
//        }).observeOn(Schedulers.io()
//        ).doOnNext(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Exception {
//                Log.d(TAG, "After observeOn(io), current thread is : " + Thread.currentThread().getName());
//            }
//        }).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Exception {
//                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
//                Log.d(TAG, "onNext: " + integer);
//            }
//        });

//        readAllRecords().subscribe(new Consumer<List<AlphabeticIndex.Record>>() {
//            @Override
//            public void accept(@NonNull List<AlphabeticIndex.Record> records) throws Exception {
//                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
//                Log.d(TAG, "onNext: " + records);
//            }
//        });

        // -----------------------------------------OKhttp知识科普----------------------------------
//        // get请求
//        // step 1: 创建 OkHttpClient 对象
//        OkHttpClient okHttpClient = new OkHttpClient();
//
//        // step 2： 创建一个请求，不指定请求方法时默认是GET。
//        Request.Builder requestBuilder = new Request.Builder().url("http://api.jisuapi.com/weather/query?appkey="+ APP_KEY +"&city=%E5%AE%89%E9%A1%BA");
//        //可以省略，默认是GET请求
//        requestBuilder.method("GET",null);
//
//        // step 3：创建 Call 对象
//        call = okHttpClient.newCall(requestBuilder.build());
//
//        //step 4: 开始异步请求
//        call.enqueue(new Callback() {
//            @Override
//            public void onFailure(Call call, IOException e) {
//                Log.d(TAG, "onFailure: " + e.toString());
//            }
//
//            @Override
//            public void onResponse(Call call, Response response) throws IOException {
//                //获得返回体
//                ResponseBody body = response.body();
//                String s = body.string();
//                Log.d(TAG, "onResponse: " + s );
//            }
//        });

        // post请求
//        //step 1: 同样的需要创建一个OkHttpClick对象
//        OkHttpClient okHttpClient = new OkHttpClient();
//
//        //step 2: 创建  FormBody.Builder
//        FormBody formBody = new FormBody.Builder()
//                .add("city", "广州")
//                .add("cityid", "")
//                .add("citycode", "")
//                .add("location", "")
//                .add("ip", "")
//                .build();
//
//        //step 3: 创建请求
//        Request request = new Request.Builder().url("http://api.jisuapi.com/weather/query?appkey="+ APP_KEY)
//                .post(formBody)
//                .build();
//
//        //step 4： 建立联系 创建Call对象
//        okHttpClient.newCall(request).enqueue(new Callback() {
//            @Override
//            public void onFailure(Call call, IOException e) {
//                Log.d(TAG, "onFailure: " + e.toString());
//            }
//
//            @Override
//            public void onResponse(Call call, Response response) throws IOException {
//                //获得返回体
//                ResponseBody body = response.body();
//                String s = body.string();
//                Log.d(TAG, "onResponse: " + s);
//            }
//        });

        // post上传
//        // step 1: 创建 OkHttpClient 对象
//        OkHttpClient okHttpClient = new OkHttpClient();
//
//        //step 2:创建 RequestBody 以及所需的参数
//        //2.1 获取文件
//        File file = new File(Environment.getExternalStorageDirectory() + "test.txt");
//        //2.2 创建 MediaType 设置上传文件类型
//        MediaType MEDIATYPE = MediaType.parse("text/plain; charset=utf-8");
//        //2.3 获取请求体
//        RequestBody requestBody = RequestBody.create(MEDIATYPE, file);
//
//        //step 3：创建请求
//        Request request = new Request.Builder().url("http://www.baidu.com")
//                .post(requestBody)
//                .build();
//
//        //step 4 建立联系
//        okHttpClient.newCall(request).enqueue(new Callback() {
//            @Override
//            public void onFailure(Call call, IOException e) {
//                // TODO: 17-1-4  请求失败
//            }
//
//            @Override
//            public void onResponse(Call call, Response response) throws IOException {
//                // TODO: 17-1-4 请求成功
//            }
//        });


        // -----------------------------------------Retrofit框架基础使用-----------------------------------------
//        //获取实例
//        Retrofit retrofit = new Retrofit.Builder()
//                //设置OKHttpClient,如果不设置会提供一个默认的
//                .client(new OkHttpClient())
//                //设置baseUrl
//                .baseUrl("https://api.github.com/")
//                //添加Gson转换器
//                .addConverterFactory(GsonConverterFactory.create())
//                .build();
//
//        GitHubService service = retrofit.create(GitHubService.class);
//
//        //同步请求
//        //https://api.github.com/users/octocat/repos
//        Call<List<Repo>> call = service.listRepos("octocat");
//        try {
//            Response<List<Repo>> repos = call.execute();
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//        //call只能调用一次。否则会抛 IllegalStateException
//        Call<List<Repo>> clone = call.clone();
//
//        Call<List<Repo>> call = service.listRepos("octocat");
//        //异步请求
//        call.enqueue(new Callback<List<Repo>>() {
//            @Override
//            public void onResponse(Call<List<Repo>> call, Response<List<Repo>> response) {
//                // Get result bean from response.body()
//                List<Repo> repos = response.body();
//                // Get header item from response
//                String links = response.headers().get("Link");
//                /**
//                 * 不同于retrofit1 可以同时操作序列化数据javabean和header
//                 */
//            }
//
//            @Override
//            public void onFailure(Call<List<Repo>> call, Throwable t) {
//
//            }
//        });

        // http://www.jianshu.com/p/128e662906af给初学者的RxJava2.0教程(三)
        // -----------------------------------------回归正题  操作符讲解 Map-----------------------------------------
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                emitter.onNext(1);
//                emitter.onNext(2);
//                emitter.onNext(3);
//            }
//        }).map(new Function<Integer, String>() {
//            // map是RxJava中最简单的一个变换操作符了, 它的作用就是对上游发送的每一个事件应用一个函数, 使得每一个事件都按照指定的函数去变化. 用事件图表示如下:
//            @Override
//            public String apply(Integer integer) throws Exception {
//                return "This is result " + integer;// int变string
//            }
//        }).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Exception {
//                Log.d(TAG, s);
//            }
//        });

        // 操作符讲解 FlatMap
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                emitter.onNext(1);
//                emitter.onNext(2);
//                emitter.onNext(3);
//            }
//        }).flatMap(new Function<Integer, ObservableSource<String>>() {
//            @Override
//            public ObservableSource<String> apply(Integer integer) throws Exception {
//                final List<String> list = new ArrayList<>();
//                for (int i = 0; i < 3; i++) {
//                    list.add("I am value " + integer);
//                }
//                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
//            }
//        }).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Exception {
//                Log.d(TAG, s);
//            }
//        });

        // 操作符讲解 concatMap
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                emitter.onNext(1);
//                emitter.onNext(2);
//                emitter.onNext(3);
//            }
//            // 这里也简单说一下concatMap吧, 它和flatMap的作用几乎一模一样, 只是它的结果是严格按照上游发送的顺序来发送的, 来看个代码吧:
//        }).concatMap(new Function<Integer, ObservableSource<String>>() {
//            @Override
//            public ObservableSource<String> apply(Integer integer) throws Exception {
//                final List<String> list = new ArrayList<>();
//                for (int i = 0; i < 3; i++) {
//                    list.add("I am value " + integer);
//                }
//                return Observable.fromIterable(list).delay(10,TimeUnit.MILLISECONDS);
//            }
//        }).subscribe(new Consumer<String>() {
//            @Override
//            public void accept(String s) throws Exception {
//                Log.d(TAG, s);
//            }
//        });


//        api.register(new RegisterRequest())            //发起注册请求
//                .subscribeOn(Schedulers.io())               //在IO线程进行网络请求
//                .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求注册结果
//                .doOnNext(new Consumer<RegisterResponse>() {
//                    @Override
//                    public void accept(RegisterResponse registerResponse) throws Exception {
//                        //先根据注册的响应结果去做一些操作
//                    }
//                })
//                .observeOn(Schedulers.io())                 //回到IO线程去发起登录请求
//                .flatMap(new Function<RegisterResponse, ObservableSource<LoginResponse>>() {
//                    @Override
//                    public ObservableSource<LoginResponse> apply(RegisterResponse registerResponse) throws Exception {
//                        return api.login(new LoginRequest());
//                    }
//                })
//                .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求登录的结果
//                .subscribe(new Consumer<LoginResponse>() {
//                    @Override
//                    public void accept(LoginResponse loginResponse) throws Exception {
//                        Toast.makeText(MainActivity.this, "登录成功", Toast.LENGTH_SHORT).show();
//                    }
//                }, new Consumer<Throwable>() {
//                    @Override
//                    public void accept(Throwable throwable) throws Exception {
//                        Toast.makeText(MainActivity.this, "登录失败", Toast.LENGTH_SHORT).show();
//                    }
//                });


        // -----------------------------------------rxbinding框架使用-----------------------------------------
//        // 防止Button短时间内多次点击代码
//        RxView.clicks(bt_content)
//                //   .sample(1,TimeUnit.SECONDS)
//                .throttleFirst(3, TimeUnit.SECONDS)//在一秒内只取第一次点击
//                .subscribe(new Consumer<Object>() {
//                    @Override
//                    public void accept(Object o) throws Exception {
//
//                        Toast.makeText(MainActivity.this, "狗子" + Thread.currentThread().getName(), Toast.LENGTH_SHORT).show();
//                    }
//                });

        // -----------------------------------------http://www.jianshu.com/p/bb58571cdb64   给初学者的RxJava2.0教程(四)// -----------------------------------------
        // Zip操作符
//        Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
//                Log.d(TAG, "emit 1");
//                e.onNext(1);
////                Thread.sleep(1000);
//
//                Log.d(TAG, "emit 2");
//                e.onNext(2);
////                Thread.sleep(1000);
//
//                Log.d(TAG, "emit 3");
//                e.onNext(3);
////                Thread.sleep(1000);
//
//                Log.d(TAG, "emit 4");
//                e.onNext(4);
////                Thread.sleep(1000);
//
//                Log.d(TAG, "emit complete1");
//                e.onComplete();
//            }
//        }).subscribeOn(Schedulers.io());
//
//        Observable<String> stringObservable = Observable.create(new ObservableOnSubscribe<String>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
//                Log.d(TAG, "emit A");
//                e.onNext("A");
//                Thread.sleep(1000);
//
//                Log.d(TAG, "emit B");
//                e.onNext("B");
//                Thread.sleep(1000);
//
//                Log.d(TAG, "emit C");
//                e.onNext("C");
//                Thread.sleep(1000);
//
//                Log.d(TAG, "emit complete2");
//                e.onComplete();
//
//            }
//        }).subscribeOn(Schedulers.io());
//
//        Observable.zip(integerObservable, stringObservable, new BiFunction<Integer, String, String>() {
//            @Override
//            public String apply(@NonNull Integer integer, @NonNull String s) throws Exception {
//                return integer + s;
//            }
//        }).subscribe(new Observer<String>() {
//            @Override
//            public void onSubscribe(@NonNull Disposable d) {
//                Log.d(TAG, "onSubscribe");
//            }
//
//            @Override
//            public void onNext(@NonNull String s) {
//                Log.d(TAG, "onNext: " + s);
//            }
//
//            @Override
//            public void onError(@NonNull Throwable e) {
//                Log.d(TAG, "onError");
//            }
//
//            @Override
//            public void onComplete() {
//                Log.d(TAG, "onComplete");
//            }
//        });

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {   //无限循环发事件
                    emitter.onNext(i);
                }
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Thread.sleep(2000);
                        Log.d(TAG, "" + integer);
                    }
                });

    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.bt_content:
//        try {
//            // 当然 Get 也支持阻塞方式的同步请求，不过在开发中这种方法很少被使用。上面我们也说了Call有一个 execute() 方法，
//            // 你也可以直接调用 call.execute() 返回一个 Response 。然后利用 isSuccessful() 判读是否成功,进行相应的结果解析。
//            Response execute = call.execute();
//            boolean successful = execute.isSuccessful();
//            if (successful) {
//                Log.d(TAG, "请求成功");
//                String string = execute.body().string();
//            } else {
//                Log.d(TAG, "请求失败");
//            }
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
                break;
        }
    }

    // 读取数据库耗时操作
    public Observable<List<AlphabeticIndex.Record>> readAllRecords() {
        return Observable.create(new ObservableOnSubscribe<List<AlphabeticIndex.Record>>() {
            @Override
            public void subscribe(ObservableEmitter<List<AlphabeticIndex.Record>> emitter) throws Exception {
                Cursor cursor = null;
                try {
//                    cursor = getReadableDatabase().rawQuery("select * from " + TABLE_NAME, new String[]{});
                    List<AlphabeticIndex.Record> result = new ArrayList<>();
                    while (cursor.moveToNext()) {
//                        result.add(Db.Record.read(cursor));
                    }
                    emitter.onNext(result);
                    emitter.onComplete();
                } finally {
                    if (cursor != null) {
                        cursor.close();
                    }
                }
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }


}
